package mrunit;

import com.ailk.oci.ocnosql.common.rowkeygenerator.GenRKCallBackHBImpl;
import com.ailk.oci.ocnosql.common.rowkeygenerator.MD5RowKeyGenerator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.IdentityTableReducer;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by zhaopan on 2017/6/29.<br>
 *
 * @Description:
 */
public class PutIndex {
    public static class TokenizerMapper
            extends Mapper<LongWritable, Text, Text, Put> {
        private Text callenumWord = new Text();
        private Text callImeiWord = new Text();
        private Text calledImeiWord = new Text();
        private final static List<String> columns = new ArrayList<String>();


        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] wordArray = line.split(",");
            StringBuffer sb = new StringBuffer();
            if (wordArray.length == 37) {
                String beginTime = wordArray[0];
                String phone = wordArray[18];
                String callednum = wordArray[19];
                String callImei = wordArray[22];
                String calledImei = wordArray[23];
                if (phone != null && beginTime != null && !"".equals(phone) && !"".equals(beginTime)) {
                    //主表rowkey
                    String rowKeyMd5 = new MD5RowKeyGenerator().generatePrefix(phone).toString();
                    String rowKey = new GenRKCallBackHBImpl().callback(rowKeyMd5 + "" + phone + "" + beginTime, line, null);//
                    byte[] family = Bytes.toBytes("0");//列族
                    byte[] qualifier = Bytes.toBytes("IX");//列
                    if (callednum != null && !"".equals(callednum)) {
                        //从表被叫rowkey
                        String callenumIndexMd5 = new MD5RowKeyGenerator().generatePrefix(callednum).toString();
                        String callenumIndexRowKey = new GenRKCallBackHBImpl().callback(callenumIndexMd5 + "!" + callednum + "" + beginTime, line, null);//
                        callenumWord.set(callenumIndexRowKey);
                        Put put = new Put(Bytes.toBytes(callenumIndexRowKey));
                        put.addColumn(family, qualifier, Bytes.toBytes(rowKey));
                        context.write(callenumWord, put);
                    }
                    if (callImei != null && !"".equals(callImei)) {
                        //从表IMEIrowkey
                        String callImeiIndexMd5 = new MD5RowKeyGenerator().generatePrefix(callImei).toString();
                        String callImeiIndexRowKey = new GenRKCallBackHBImpl().callback(callImeiIndexMd5 + "@" + callImei + "" + beginTime, line, null);//
                        callImeiWord.set(callImeiIndexRowKey);
                        Put put = new Put(Bytes.toBytes(callImeiIndexRowKey));
                        put.addColumn(family, qualifier, Bytes.toBytes(rowKey));
                        context.write(callImeiWord, put);
                    }
                    if (calledImei != null && !"".equals(calledImei)) {
                        //从表IMEIrowkey
                        String calledImeiIndexMd5 = new MD5RowKeyGenerator().generatePrefix(calledImei).toString();
                        String calledImeiIndexRowKey = new GenRKCallBackHBImpl().callback(calledImeiIndexMd5 + "#" + calledImei + "" + beginTime, line, null);//
                        calledImeiWord.set(calledImeiIndexRowKey);
                        Put put = new Put(Bytes.toBytes(calledImeiIndexRowKey));
                        put.addColumn(family, qualifier, Bytes.toBytes(rowKey));
                        context.write(calledImeiWord, put);
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.security.authentication", "kerberos");
        conf.set("hbase.zookeeper.quorum", "hbbdch-dn-01:2181,hbbdch-nn-01:2181,hbbdch-nn-02:2181");
        conf.set("zookeeper.znode.parent", "/services/slider/users/client/hbase_1007_3");
        conf.set("hbase.master.kerberos.principal", "client/_HOST@HBBDCH.COM");
        conf.set("hbase.regionserver.kerberos.principal", "client/_HOST@HBBDCH.COM");
        conf.setInt("hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily", 5000);
        conf.set("hbase.fs.tmp.dir", "/tmp/hbase-staging");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        Job job = Job.getInstance(conf, "create index to hbase");
        job.setJarByClass(PutIndex.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Put.class);//important
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        TableMapReduceUtil.initTableReducerJob(otherArgs[1], IdentityTableReducer.class, job);
        job.setNumReduceTasks(0);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}